#include "stdafx.h"

/************************************************************************/
/* File: MultiThreadEventManager.cpp	Author: BG Studio, Justin Lee	*/
/* Description:									                        */
/* Implements of CMultiThreadEventManager and CMultiThreadEvent.	    */
/* We maintain an event queue to manage events:			                */
/* If there are events in queue our work thread will pick it up and		*/
/* reacts, then notify the waiting thread or directly pick up next one.	*/
/* If there is not event in queue, the work thread blocks until some	*/
/* events enter this queue.									            */
/* Tested both on Windows and Linux.                                    */
/************************************************************************/
#if defined(WIN32)
#include <Windows.h>
#pragma comment(lib,"libs/pthreads/pthreadVC2.lib")
#else
#include <unistd.h>
#endif

#include "MultiThreadEventManager.h"

using namespace std;

/************************************************************************/
/* CMultiThreadEvent references                                         */
/************************************************************************/
CMultiThreadEvent::CMultiThreadEvent()
	: is_in_processed(false)
	, need_wait(true)
	, succeed_processed(false)
	, p_event_lock(NULL)
	, p_event_count(NULL)
	, n_count(0)
{
	sem_init(&finished,0,0);
}

CMultiThreadEvent::~CMultiThreadEvent()
{
	sem_destroy(&finished);
}

bool CMultiThreadEvent::Wait()
{
	if (!need_wait)
	{
		return true;
	}
	sem_wait(&finished);
	pthread_mutex_lock(p_event_lock);
	is_in_processed=false;
	bool rtval=succeed_processed;
	--n_count;
	if (n_count>0)
	{
		sem_post(p_event_count);
	}
	pthread_mutex_unlock(p_event_lock);
	return rtval;
}

/************************************************************************/
/* CMultiThreadEventManager references                                  */
/************************************************************************/
CMultiThreadEventManager::CMultiThreadEventManager(void)
	: n_threads(0)
	, end_wait_all(true)
{
	pthread_mutex_init(&event_lock,NULL);
	sem_init(&event_count,0,0);

	int nCPU=0;
	//Get number of CPUs
#if defined(WIN32)
	nCPU=pthread_num_processors_np();
#else
	nCPU=(int)sysconf(_SC_NPROCESSORS_ONLN);
#endif
	if(nCPU>0)
	{
		n_threads=nCPU<<1;//2 threads per processor
		n_threads=PTHREAD_THREADS_MAX>n_threads?n_threads:PTHREAD_THREADS_MAX;
	}
	else
	{
		//symbian will cause bug of returning -1 here,
		//so we use only two thread for switching.
		n_threads=1;
	}
	is_run=true;
	//create threads
	for (int i=0;i<n_threads;++i)
	{
		pthread_create(&threads[i],NULL,WorkThread,(void*)this);

		//Set threads affinities
#if defined(WIN32)
		HANDLE hT=pthread_getw32threadhandle_np(threads[i]);
		//Use default win32 API for ideal processor rather than forceful affinity
		DWORD dwrt=SetThreadIdealProcessor(hT,(DWORD)(i>>1));
#elif defined(__GCCE__)|defined(__WINSCW__)
		//None
#else
		//Directly affinities
		cpu_set_t cpu_info;
		CPU_ZERO(&cpu_info);
		CPU_SET(i>>1, &cpu_info);
		pthread_setaffinity_np(threads[i],sizeof(cpu_info),&cpu_info);
#endif
	}
}

CMultiThreadEventManager::~CMultiThreadEventManager(void)
{
	//signal all threads
	is_run=false;
	//prevent thread blocks in waiting state
	pthread_mutex_lock(&event_lock);
	if (event_queue.empty())
	{
		for (int i=0;i<n_threads;++i)
		{
			sem_post(&event_count);
		}
	}
	pthread_mutex_unlock(&event_lock);
	//wait for all threads end
	for (int i=0;i<n_threads;++i)
	{
		void* pvrt=NULL;
		pthread_join(threads[i],&pvrt);
	}
	//force all the waiting event notified
	deque<CMultiThreadEvent*>::iterator ie;
	for (ie=event_queue.begin();ie!=event_queue.end();++ie)
	{
		CMultiThreadEvent* pe=*ie;
		if (pe->need_wait)
		{
			pthread_mutex_lock(&event_lock);
			pe->succeed_processed=false;
			pthread_mutex_unlock(&event_lock);
			sem_post(&pe->finished);
		}
	}
	//release resource
	event_queue.clear();
	pthread_mutex_destroy(&event_lock);
	sem_destroy(&event_count);
}

bool CMultiThreadEventManager::SignalEvent(CMultiThreadEvent* pEvent)
{
	if (!is_run)
	{
		return false;
	}
	pthread_mutex_lock(&event_lock);
	pEvent->p_event_count=&event_count;
	pEvent->p_event_lock=&event_lock;
	event_queue.push_back(pEvent);
	//signal count increased
	++pEvent->n_count;
	//No repeat, then increase semaphore
	if(pEvent->n_count<=1)
	{
		sem_post(&event_count);
	}
	pthread_mutex_unlock(&event_lock);
	return true;
}

void* CMultiThreadEventManager::WorkThread(void* obj)
{
	CMultiThreadEventManager* pMgr=(CMultiThreadEventManager*)obj;
	pMgr->DoWorkThread();
	pthread_exit(NULL);
	return NULL;
}

void CMultiThreadEventManager::DoWorkThread()
{
	while (is_run||end_wait_all)
	{
		pthread_mutex_lock(&event_lock);
		if (end_wait_all&&event_queue.empty()&&!is_run)
		{
			//prevent thread blocks in waiting state
			for (int i=0;i<n_threads;++i)
			{
				sem_post(&event_count);
			}
			pthread_mutex_unlock(&event_lock);
			break;
		}
		pthread_mutex_unlock(&event_lock);
		//Decrease counter
		sem_wait(&event_count);
		pthread_mutex_lock(&event_lock);
		//Pick up event from queue
		CMultiThreadEvent* pe=NULL;
		deque<CMultiThreadEvent*>::iterator ie=event_queue.begin();
		while (ie!=event_queue.end()&&(*ie)->is_in_processed)
		{
			++ie;
		}
		if (ie!=event_queue.end())
		{
			pe=*ie;
			event_queue.erase(ie);
			pe->is_in_processed=true;
		}
		else
		{
			//No event have been picked up, so recover it.
			sem_post(&event_count);
		}
		pthread_mutex_unlock(&event_lock);

		if (pe)
		{
			pe->OnEvent(this);

			pthread_mutex_lock(&event_lock);
			if (!pe->need_wait)
			{
				pe->is_in_processed=false;
				--pe->n_count;
				if (pe->n_count>0)
				{
					sem_post(&event_count);
				}
			}
			else
			{
				//signal the waiting event.
				pe->succeed_processed=true;
				sem_post(&pe->finished);
			}
			pthread_mutex_unlock(&event_lock);
		}
	}
}
